package org.zhq.impl.async;

import org.zhq.core.IoArgs;
import org.zhq.core.ReceiveDispatcher;
import org.zhq.core.ReceivePacket;
import org.zhq.core.Receiver;
import org.zhq.util.CloseUtils;

import java.util.concurrent.atomic.AtomicBoolean;

/**
 * 异步接收调度器
 *
 * @author zhengquan
 * @date 2019/2/2
 */
public class AsyncReceiveDispatcher implements ReceiveDispatcher, IoArgs.IoArgsEventProcessor, AsyncPacketWriter.PacketProvider {

    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final Receiver receiver;
    private final ReceivePacketCallback callback;

    private final AsyncPacketWriter writer = new AsyncPacketWriter(this);

    public AsyncReceiveDispatcher(Receiver receiver, ReceivePacketCallback callback) {
        this.receiver = receiver;
        this.callback = callback;
    }

    @Override
    public void start() {
        receiver.setReceiveListener(this);
        registerReceive();
    }

    @Override
    public void stop() {
        receiver.setReceiveListener(null);
    }

    @Override
    public void close() {
        if (closed.compareAndSet(false, true)) {
            writer.close();
            receiver.setReceiveListener(null);
        }
    }

    /**==============================IoArgs.IoArgsEventProcessor==================================*/

    @Override
    public IoArgs provideIoArgs() {
        // 一份新的IoArgs需要调用一次开始写入数据的操作
        IoArgs args = writer.takeIoArgs();
        args.startWriting();
        return args;
    }

    @Override
    public boolean onConsumeFailed(Throwable e) {
        CloseUtils.close(this);
        return true;
    }

    @Override
    public boolean onConsumeCompleted(IoArgs args) {
        final AtomicBoolean closed = this.closed;
        final AsyncPacketWriter writer = this.writer;

        args.finishWriting();
        do {
            writer.consumeIoArgs(args);
        }while(args.remained() && !closed.get());
        registerReceive();

        return !closed.get();
    }


    /**===========================================================================================*/

    /**============================AsyncPacketWriter.PacketProvider===============================*/

    /**
     *
     * @param type
     * @param length
     * @param headerInfo
     * @return
     */
    @Override
    public ReceivePacket takePacket(byte type, long length, byte[] headerInfo) {
        return callback.onArrivedNewPacket(type, length,headerInfo);
    }

    /**
     * 接收包完成的回调
     *
     * @param packet
     * @param succeed
     */
    @Override
    public void completedPacket(ReceivePacket packet, boolean succeed) {
        CloseUtils.close(packet);
        callback.onReceivedPacketCompleted(packet);
    }

    @Override
    public void onReceivedHeartbeat() {
        callback.onReceivedHeartbeat();
    }

    /**===========================================================================================*/

    /**
     * 请求网络进行数据接收
     */
    private void registerReceive() {
        try {
            receiver.postReceiveAsync();
        } catch (Exception e) {
            e.printStackTrace();
            CloseUtils.close(this);
        }
    }

}
